package rpc

import (
	"bytes"
	"fmt"
	"github.com/panjf2000/gnet"
	"github.com/panjf2000/gnet/pkg/logging"
	"log"
	"sync/atomic"
	"time"
)

type HttpServer struct {
	*gnet.EventServer
	eng          gnet.Server
	network      string
	addr         string
	connected    int32
	disconnected int32
	router       HttpRouter
}

func NewHttpServerV1(addr string, router HttpRouter) *HttpServer {
	return &HttpServer{
		network:      "tcp",
		addr:         addr,
		connected:    0,
		disconnected: 0,
		router:       router,
	}
}

func (s *HttpServer) OnInitComplete(srv gnet.Server) (action gnet.Action) {
	logging.Infof("running server on %s", fmt.Sprintf("%s://%s", s.network, s.addr))
	s.eng = srv
	return
}

func (s *HttpServer) OnOpened(srv gnet.Conn) (out []byte, action gnet.Action) {
	code := HttpCodecPool.Get().(*HttpCodec)
	code.Reset()
	srv.SetContext(code)
	atomic.AddInt32(&s.connected, 1)
	return
}

func (s *HttpServer) OnClosed(c gnet.Conn, err error) (action gnet.Action) {
	if err != nil {
		log.Printf("error occurred on connection=%s, %v\n", c.RemoteAddr().String(), err)
	}
	atomic.AddInt32(&s.connected, -1)
	HttpCodecPool.Put(c.Context())
	return
}

func (s *HttpServer) React(frame []byte, c gnet.Conn) (out []byte, action gnet.Action) {
	hc := c.Context().(*HttpCodec)
	if len(frame) == 0 {
		return
	}
	reader := bytes.NewReader(frame)
	if hc.Finish {
		n, err := reader.Read(hc.TmpBuf[0:])
		if err != nil {
			hc.Reset()
			return
		}
		hc.Head.Write(hc.TmpBuf[:n])
		headerOffset, miss := hc.Parser.Parse(hc.Head.Bytes()[:n])
		for miss == ErrMissingData {
			m, read := reader.Read(hc.TmpBuf[0:])
			if read != nil {
				hc.Reset()
				return
			}
			n += m
			hc.Head.Write(hc.TmpBuf[:m])
			headerOffset, miss = hc.Parser.Parse(hc.Head.Bytes()[:n])
		}
		if miss != nil && miss != ErrMissingData {
			hc.Reset()
			action = gnet.Close
			return
		}
		hc.Body.Write(hc.Head.Bytes()[headerOffset:n])
	}
	err := s.fetchBody(reader, hc)
	if err != nil {
		hc.Reset()
		return
	}
	if hc.Finish {
		s.router.Publish(hc, hc.Body.Bytes())
		out = hc.OutBuf.Bytes()
		hc.Reset()
	}
	return
}

func (s *HttpServer) fetchBody(c *bytes.Reader, hc *HttpCodec) error {
	for hc.Body.Len() < hc.Parser.ContentLength() {
		n, _ := c.Read(hc.TmpBuf[0:])
		if n == 0 {
			break
		}
		hc.Body.Write(hc.TmpBuf[:n])
	}
	if hc.Body.Len() < hc.Parser.ContentLength() {
		hc.Finish = false
	} else if hc.Body.Len() == hc.Parser.ContentLength() {
		hc.Finish = true
	} else {
		panic("message error")
	}
	return nil
}

func (s *HttpServer) Start() {
	err := gnet.Serve(s, s.addr,
		gnet.WithMulticore(true),
		gnet.WithTCPKeepAlive(time.Minute*5),
		gnet.WithReusePort(true),
		gnet.WithSocketRecvBuffer(8*1024),
		gnet.WithSocketSendBuffer(8*1024),
		gnet.WithTCPNoDelay(gnet.TCPNoDelay),
	)
	if err != nil {
		logging.Errorf("start server error", err)
	}
}
